
#include <ttg/all.h>

using namespace ttg::threads;
using namespace ttg::services;
using namespace ttg::pipeline;

//Auxiliary structure, that keeps rates of each literals.
struct LiteralRates
{
	public:
		ttg::threads::AtomicInt literals[256];
		
		//Creates zero rates.
		inline LiteralRates()
		{ clear(); }
		//Clears all rates.
		inline void clear()
		{
			for (size_t i = 0; i < 256; i++)
				literals[i] = 0;
		}
};

//Auxiliary structure, that keeps portion of file and provides access to its data.
struct Buffer
{
	private:
		char *data;
		size_t max_size;
		size_t cur_size;
	public:
		//Creates uninitialized buffer with requred size.
		inline Buffer(size_t max_size)
		{
			this->max_size = max_size;
			this->cur_size = 0;
			this->data = new char[max_size];
		}
		//Returns current size of buffer.
		inline size_t getSize()
		{ return cur_size; }
		//Returns maximum size, that can be setted for buffer.
		inline size_t getMaxSize()
		{ return max_size; }
		//Sets new actual size for buffer.
		inline void setSize(size_t cur_size)
		{ this->cur_size = cur_size <= max_size ? cur_size : max_size; }
		//Returns raw data.
		inline char *getData()
		{ return data; }
		//Releases allocated memory.
		~Buffer()
		{ SAFE_DELETE(data); }
};

//Counts rates of letter, numeric and punctuation literals for incoming buffer.
//Note: received buffer will be deleted by this node and after that
//		user variable "buffer_count" will be decreased by one.
class RateCounter :public Node<Buffer *, bool>
{
	private:
		LiteralRates *rates;						//reference to variable, defined in user context.
		ttg::threads::AtomicInt *buffer_count;		//reference to variable, defined in user context.
	public:
		virtual void process(Buffer *data)
		{
			//Calculating rates of letters.
			//Note: we can forget about potential data race due to variable atomicity.
			for (size_t i = 0; i < data->getSize(); i++)
			{
				char val = data->getData()[i];
				if ('a' <= val && val <= 'z' ||
					'A' <= val && val <= 'Z' ||
					'0' <= val && val <= '9' ||
					val == '.' || val == ',' ||
					val == '!' || val ==  '?')
					rates->literals[val].inc();
			}
			//Deleting buffer and decreasing count of buffers.
			SAFE_DELETE(data);
			buffer_count->dec();
		}
		//Notifies, that structure of the pipeline has been changed.
		virtual void onModification()
		{
			//Updating references to context variables.
			rates = &getContext().getVariable<LiteralRates>("rates");
			buffer_count = &getContext().getVariable<AtomicInt>("buffer_count");
		}
};

int main()
{
	//Starting services and creating pipeline.
	ttg::services::ServiceBase::getRef()->start();
	//ttg::services::Services s(true);
	Pipeline<Buffer *, bool> pl(wrap(new RateCounter()));
	//Creating user context for pipeline.
	LiteralRates rates;
	AtomicInt buffer_count;
	UserContext context;
	context.registerVariable("rates", &rates);
	context.registerVariable("buffer_count", &buffer_count);
	pl.setUserContext(context);									//now "rates" and "buffer_count" variables
																//can be accessed from any node of pipeline.
	//Opening file with a big text (that is provided with this sample).
	FILE *f = fopen("CriminalCode.txt", "r");
	if (f == NULL)
	{
		ttg::services::Logger::getRef()->addError("Cannot find file");
		return 0;
	}
	//Loading data and starting processing.
	Buffer *b;
	do
	{
		//Guard, that avoids memory overflow.
		while (buffer_count.get() >= (int)CPUAnalyzer::getRef()->getCoreCount())
			ttg::threads::CurrentThread::sleep(0.02);
		//Loading another portion of data.
		b = new Buffer(64 * 1024);
		b->setSize(fread(b->getData(), sizeof(char), b->getMaxSize(), f));
		buffer_count.inc();
		pl.start(b);					//note: it's asynchronous call - continuing data loading
	}
	while (b->getSize() != 0);			//is it EOF?
	//Finishing data processing and printing rates.
	pl.finish();
	for (size_t i = 0; i < 256; i++)
		if (rates.literals[i].get() != 0)
			printf("%c: %d\n", (unsigned char)i, rates.literals[i].get());
	exit(0);
}
